package com.atguigu.cdc;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

/**
 * 自定义反序列化器。将 flinkCDC 输出的字符串格式转换成类似于 maxwell 输出的 json 字符串
 *
 * @author lvbingbing
 * @date 2022-03-19 13:07
 */
public class FlinkCDCMaxwellDeserializer implements DebeziumDeserializationSchema<String> {

    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        Struct valueStruct = (Struct) sourceRecord.value();
        Struct sourceStruct = valueStruct.getStruct("source");
        JSONObject jsonObj = new JSONObject();
        // 获取数据库名称
        String database = sourceStruct.getString("db");
        jsonObj.put("database", database);
        // 获取表名
        String table = sourceStruct.getString("table");
        jsonObj.put("table", table);
        // 获取操作类型
        String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();
        if (type.equals("create")) {
            type = "insert";
        }
        jsonObj.put("type", type);
        // 封装 data
        Struct afterStruct = valueStruct.getStruct("after");
        JSONObject jsonObject = new JSONObject();
        if (afterStruct != null) {
            List<Field> fields = afterStruct.schema().fields();
            for (Field field : fields) {
                String fieldName = field.name();
                Object fieldValue = afterStruct.get(fieldName);
                jsonObject.put(fieldName, fieldValue);
            }
        }
        jsonObj.put("data", jsonObject);
        // 收集数据
        collector.collect(jsonObj.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }

    /**
     * 测试
     *
     * @param args <br>
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall2021_realtime")
                .tableList("gmall2021_realtime.t_user")
                .username("root")
                .password("139559")
                .startupOptions(StartupOptions.initial())
                .deserializer(new FlinkCDCMaxwellDeserializer())
                .build();
        env.addSource(sourceFunction).print();
        env.execute();
    }
}
